package com.xdxc.service.impl;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xdxc.component.registry.ValidationRegistry;
import com.xdxc.component.validation.AnswerSanitizer;
import com.xdxc.component.validation.AnswerValidator;
import com.xdxc.mapper.UserAnswerMapper;
import com.xdxc.model.dto.UserAnswerDto;
import com.xdxc.model.entity.UserAnswer;
import com.xdxc.response.Result;
import com.xdxc.service.UserAnswerService;
import com.xdxc.utils.SecurityUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;

import java.util.*;

import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class UserAnswerServiceImpl extends ServiceImpl<UserAnswerMapper, UserAnswer> implements UserAnswerService {

    @Autowired
    private  RedisService redisService;

    @Autowired
    private ValidationRegistry validationRegistry;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private AnswerSanitizer answerSanitizer;
    @Override
    public Result<String> saveUserAnswer(Long paperId, UserAnswerDto userAnswerDto) {
        // 参数校验
        if (paperId == null || userAnswerDto == null) {
            return Result.error("参数不能为空");
        }
        Long userId = SecurityUtils.getCurrentUserId();
        if (userId == null) {
            return Result.error("用户未登录");
        }
        AnswerValidator validator = validationRegistry.getValidator(userAnswerDto.getAnswerType());
        if (validator == null) {
            return Result.error("未知题目类型");
        }
        if (!validator.validate(userAnswerDto)) {
            return Result.error("答案格式错误");
        }


        //清洗
        UserAnswerDto  userAnswerDto1  =answerSanitizer.sanitize(userAnswerDto);
        // 构建持久化对象
        UserAnswer userAnswer = buildUserAnswer(userId, userAnswerDto1);
        // 设置消息唯一标识
        userAnswerDto1.setUserId(userId);
        userAnswerDto1.setUuId(buildMessageKey(userAnswerDto1));
        userAnswerDto1.setExamId(userAnswerDto.getExamId());
        // 异步发送延迟消息
        sendDelayedMessage(userAnswerDto1);
        // 更新Redis缓存
        updateAnswerCache(paperId, userId,userAnswerDto,userAnswer);
        return Result.success("提交成功");
    }

    private UserAnswer buildUserAnswer(Long userId, UserAnswerDto dto) {
        return UserAnswer.builder()
                .quId(dto.getQuId())
                .answerIds(dto.getAnswerIds())
                .answerText(dto.getAnswerText())
                .build();
    }

    private void sendDelayedMessage(UserAnswerDto dto) {
        String topic = "ANSWER_SAVE_TOPIC";
        String tags = String.valueOf(dto.getAnswerType());

        // 添加调试日志
        log.debug("准备发送延迟消息，dto: {}", dto);
        dto.setBornTimestamp(System.currentTimeMillis());
        Message<UserAnswerDto> message = MessageBuilder.withPayload(dto)
                .setHeader(RocketMQHeaders.KEYS, dto.getUuId())
                // 修改为实际需要的延迟级别，例如3表示10秒延迟
                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 3)
                .build();
        try {

            SendResult result = rocketMQTemplate.syncSendOrderly(
                    "Exam:userAnswer",
                    MessageBuilder.withPayload(dto)
                            .setHeader(MessageConst.PROPERTY_KEYS, dto.getCheckId().toString()) // 设置消息Key
                            .setHeader("userAnswer", dto.getCheckId()) // 自定义Header
                            .build(),
                    dto.getCheckId().toString() // 用于顺序消息的shardingKey
            );
            log.info("消息发送结果：status={}, msgId={}", result.getSendStatus(), result.getMsgId());
            log.info("🚀 消息发送成功，checkId={}", dto.getCheckId());
        } catch (Exception e) {
            log.error("❌ 消息发送失败，checkId={}", dto.getCheckId(), e);
        }
        rocketMQTemplate.asyncSend(topic + ":" + tags, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("消息发送成功，消息ID: {}, 发送状态: {}",
                        sendResult.getMsgId(),
                        sendResult.getSendStatus());
            }
            @Override
            public void onException(Throwable e) {
                log.error("答案消息发送失败，checkId={}, uuid={}",
                        dto.getCheckId(),
                        dto.getUuId(),
                        e);
            }
        }, 3000);

        log.debug("消息发送调用完成（异步）");
    }








    private void updateAnswerCache(Long paperId, Long userId, UserAnswerDto userAnswerDto,UserAnswer userAnswer) {
        String cacheKey = buildCacheKey(paperId, userId);
        String field = String.valueOf(userAnswerDto.getCheckId());
        redisService.updateField(cacheKey, field, userAnswer);
        log.debug("Redis缓存已更新，key: {}，field: {}", cacheKey, field);
    }
    private String buildCacheKey(Long paperId, Long userId) {
        return String.format("%d_%d", paperId, userId);
    }

    private String buildMessageKey(UserAnswerDto dto) {
        // 确保字段不为空，避免 NullPointerException
        String paperId = String.valueOf(Objects.requireNonNull(dto.getPaperId(), "paperId不能为null"));
        String userId = String.valueOf(Objects.requireNonNull(dto.getUserId(), "userId不能为null"));
        String checkId = String.valueOf(Objects.requireNonNull(dto.getCheckId(), "checkId不能为null"));
        // 拼接 Key
        String key = String.join(":", paperId, userId, checkId);
        // 截取前 128 字节，确保 Key 不超长
        if (key.getBytes(StandardCharsets.UTF_8).length > 128) {
            return new String(key.getBytes(StandardCharsets.UTF_8), 0, 128, StandardCharsets.UTF_8);
        }
        return key;
    }
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void sendExamAnswers(Long examId) {
        final int pageSize = 1000; // 每批1000条
        final int maxRetry = 3; // 最大重试次数
        final int maxConcurrent = 5; // 最大并发批次
        final int rateLimit = 100; // 每秒最多发送100批次
        final RateLimiter rateLimiter = RateLimiter.create(rateLimit);
        // 使用AtomicInteger保证页码在lambda中的有效final
        final AtomicInteger currentPage = new AtomicInteger(1);
        final Semaphore semaphore = new Semaphore(maxConcurrent);
        final ExecutorService executor = Executors.newFixedThreadPool(maxConcurrent);
        final List<Future<?>> futures = new ArrayList<>();
        // 预构建固定查询条件
        final LambdaQueryWrapper<UserAnswer> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(UserAnswer::getExamId, examId);
        try {
            while (true) {
                // 获取速率限制许可
                rateLimiter.acquire();
                // 使用final局部变量保存当前页码
                final int currentPageNum = currentPage.get();
                // 分页查询考试数据
                Page<UserAnswer> userAnswerPage = new Page<>(currentPageNum, pageSize);
                IPage<UserAnswer> pagedUserAnswers = this.page(userAnswerPage, queryWrapper);
                List<UserAnswer> answerList = pagedUserAnswers.getRecords();
                if (CollectionUtils.isEmpty(answerList)) {
                    break;
                }
                // 获取并发控制许可
                semaphore.acquire();
                // 使用final局部变量传递数据到lambda
                final List<UserAnswer> finalAnswerList = answerList;
                // 提交发送任务到线程池
                Future<?> future = executor.submit(() -> {
                    try {
                        sendWithRetry(examId, currentPageNum, finalAnswerList, maxRetry);
                    } finally {
                        semaphore.release();
                    }
                });
                futures.add(future);
                // 提前检查是否最后一页
                if (currentPageNum >= pagedUserAnswers.getPages()) {
                    break;
                }
                currentPage.incrementAndGet();
            }
            // 等待所有任务完成
            for (Future<?> future : futures) {
                future.get();
            }
            log.info("完成发送所有考试答案, examId: {}, 总页数: {}", examId, currentPage.get() - 1);
        } catch (Exception e) {
            log.error("发送考试答案过程中发生异常, examId: {}", examId, e);
            throw new RuntimeException("发送考试答案失败", e);
        } finally {
            executor.shutdown();
        }
    }


    private void sendWithRetry(Long examId, int currentPage, List<UserAnswer> answerList, int maxRetry) {
        int retryCount = 0;
        while (retryCount <= maxRetry) {
            try {
                // 1. 构建真正的批量消息（每条消息包含多个答案）
                Message<List<UserAnswer>> message = MessageBuilder.withPayload(answerList)
                        .setHeader(MessageConst.PROPERTY_TAGS, "SCORING")
                        .setHeader(MessageConst.PROPERTY_KEYS, examId.toString())
                        .build();

                // 2. 批量发送（使用真正的批量消息）
                rocketMQTemplate.syncSend("EXAM_ANSWER_TOPIC", message);
                log.info("成功发送批次: {}, examId: {}, 本批次大小: {}",
                        currentPage, examId, answerList.size());
                return;
            } catch (Exception e) {
                retryCount++;
                if (retryCount > maxRetry) {
                    log.error("发送批次 {} 达到最大重试次数 {}, examId: {}",
                            currentPage, maxRetry, examId, e);
                    throw new RuntimeException("发送消息失败，达到最大重试次数", e);
                }

                log.warn("发送批次 {} 第 {} 次重试, examId: {}",
                        currentPage, retryCount, examId, e);

                // 指数退避
                try {
                    Thread.sleep((long) Math.pow(2, retryCount) * 1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("重试被中断", ie);
                }
            }
        }
    }

}
